#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#include "thread_pool.h"
#include "debug.h"
void handler(void *arg)				
{
    pthread_mutex_unlock((pthread_mutex_t *)arg);	//防止死锁，所以在这里添加解锁操作
}

/* 每一个线程池中的线程所执行的内容， arg就是线程池的地址 */
void *routine(void *arg)
{
    thread_pool *pool = (thread_pool *)arg;	//将线程池的地址存放进去pool
    struct task *p;				//缓冲指针，后期任务队列遍历的时候使用

    while(1)
    {
        pthread_cleanup_push(handler, (void *)&pool->lock); //提前登记线程被取消后需要处理的事情
        pthread_mutex_lock(&pool->lock);				    //需要操作线程池中的共有资源，所以加锁

        while(pool->waiting_tasks == 0 && !pool->shutdown)	//判断是否没有需要运行的任务
        {
            pthread_cond_wait(&pool->cond, &pool->lock);	//让线程睡眠
        }

        //判断线程池是否没有任务并且需要关闭
        if(pool->waiting_tasks == 0 && pool->shutdown == true)	
        {
            pthread_mutex_unlock(&pool->lock);	//解锁
            pthread_exit(NULL);					//退出线程
        }

        p = pool->task_list->next;			//让p登记需要运行的任务节点
        pool->task_list->next = p->next;	//将此任务节点从链表中删除
        pool->waiting_tasks--;				//将等待运行的任务队列-1

        pthread_mutex_unlock(&pool->lock);	//解锁
        pthread_cleanup_pop(0);				//解除登记取消线程之后所做的函数

        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);		//忽略线程的取消操作
        (p->task)(p->arg);											//函数调用
        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);		//重启线程取消操作

        free(p);
    }

    pthread_exit(NULL);
}

int set_max_active_threads(thread_pool* pool, unsigned int now_max_active_threads)
{
    /*  设置最大活跃线程数。
     *  返回值: 
     *      成功返回0 ，失败返回-1，
     *      非完全成功返回失败次数
     */
    void *new_ptr;
    unsigned int old_max_active_threads = pool->max_active_threads;
    if(!now_max_active_threads){ return -1; }
    if(old_max_active_threads == now_max_active_threads) { return 0; }

    // 如果 之前最大的进程数比原来的少 那么直接拓展就好
    if(old_max_active_threads < now_max_active_threads)
    {
        DEBUG_GREEN("拓展");
        // 拓展内存
        new_ptr = realloc(pool->tids, sizeof(pthread_t) * now_max_active_threads);
        if(!new_ptr) { return -1; } else { pool->tids = new_ptr; }
    }else
    {
        //若 当前 活跃线程 超过了 最大的界限 则需要回收额外的线程和地址空间
        if(pool->active_threads > now_max_active_threads)
        {
            // 先回收线程内存，再回收分配有关的内存
            DEBUG_RED("回收");
            remove_thread(pool, pool->active_threads - now_max_active_threads);
        }

        new_ptr = realloc(pool->tids, sizeof(pthread_t) * now_max_active_threads);//申请最大线程数量的ID内存
        if(!new_ptr) {  return -1; } else { pool->tids = new_ptr; }
    }

    pool->max_active_threads = now_max_active_threads;
    return 0;
}

/* 返回 一个 pool：线程池对象（单例模式）
 * max_active_threads   :  最大的线程数量 至少为1
 * threads_number       ： 伴随初始化的线程数量 
 */
thread_pool *init_pool(unsigned int max_active_threads,
                       unsigned int threads_number)
{
    // 单例模式
    static thread_pool *pool = NULL;
    int i;
    if(pool) { DEBUG_YELLOW("[Warning] Singal pool.");  return pool;  }

    pool = malloc(sizeof(thread_pool));
    if(!pool) { return NULL; }
    memset(pool, 0, sizeof(thread_pool));

    //初始化条件变量、互斥锁
    pthread_mutex_init(&pool->lock, NULL);
    pthread_cond_init(&pool->cond, NULL);

    // 动态限制最大线程数量
    pool->max_active_threads = (!max_active_threads )? 1: max_active_threads;
    pool->shutdown  = false;							//开启线程池
    pool->task_list = malloc(sizeof(struct task));	//申请任务链表
    pool->tids      = malloc(sizeof(pthread_t) *pool->max_active_threads);
    memset(pool->tids, 0, sizeof(sizeof(pthread_t) *pool->max_active_threads));
    if(pool->task_list == NULL || pool->tids == NULL)
    {
        CPERROR(1, "allocate memory error");
        goto deinit;
    }

    pool->task_list->next = NULL;	//初始化链表头，将下一个节点指向NULL
    pool->waiting_tasks = 0;		//将等待运行的任务数量置0

    // 登记当前的线程数量
    pool->active_threads = 
        (pool->max_active_threads < threads_number)? pool->max_active_threads:  threads_number;

    for(i = 0; i < pool->active_threads; i++)	//创建线程池工作线程
    {
        if(pthread_create(&((pool->tids)[i]), NULL,
                          routine, (void *)pool) != 0)
        {
            perror("create threads error");
            goto deinit;
        }else
        {
            DEBUG_GREEN("[%d]init TID:[%u]", i, (pthread_t)pool->tids[i]);
        }
    }

    return pool;
deinit :
    free(pool);  pool = NULL;  return NULL;
}


/* 投放任务到线程池
 * pool：线程池；
 * task：函数指针； 
 * arg ：传入给task函数的参数 
 * */
bool add_task(thread_pool *pool,
              void *(*task)(void *arg), void *arg)
{
    struct task *new_task = malloc(sizeof(struct task));	//新建一个任务节点
    if(new_task == NULL)
    {
        perror("allocate memory error");
        return false;
    }
    new_task->task = task;
    new_task->arg = arg;
    new_task->next = NULL;

    pthread_mutex_lock(&pool->lock);
    if(pool->waiting_tasks >= MAX_WAITING_TASKS)		//判断任务数量有没有超标
    {
        pthread_mutex_unlock(&pool->lock);

        //fprintf(stderr, "Too many tasks to load new task.\\n");	//反馈太多任务了
        free(new_task);									//释放掉刚才登记的任务节点

        return false;									//返回添加不了任务到任务链表中
    }

    struct task *tmp = pool->task_list;					//将线程池中任务链表的头节点登记到tmp
    while(tmp->next != NULL)							//将tmp指向最后的节点的位置
        tmp = tmp->next;

    tmp->next = new_task;								//将新建的任务节点插入到链表中
    pool->waiting_tasks++;								//将等待的任务数量+1

    pthread_mutex_unlock(&pool->lock);
    pthread_cond_signal(&pool->cond);					//唤醒正在睡眠中的线程

    return true;										//添加成功
}

/* 添加线程到线程池中 pool：线程池结构体地址， additional_threads:添加的线程的数量 */
int add_thread(thread_pool *pool, unsigned additional_threads)
{
    int i, actual_increment = 0;
    unsigned total_threads =
        pool->active_threads + additional_threads;	//将总数记录在这个变量中

    if(additional_threads == 0)
        return 0;
    for(i = pool->active_threads;
        i < total_threads && i < pool->max_active_threads;
        i++)
    {
        if(pthread_create(&((pool->tids)[i]),
                          NULL, routine, (void *)pool) != 0)	//新建线程
        {
            if(actual_increment == 0)   {	return -1;  }
            break;
        }
        actual_increment++; 					//记录成功添加了多少条线程到线程池中
    }

    pool->active_threads += actual_increment;	//将最后成功添加到线程池的线程总数记录线程池中
    return actual_increment;					//返回新建线程数
}

int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
    int i;
    int remain_threads = pool->active_threads - removing_threads;	//登记移除线程之后的线程数量
    remain_threads = remain_threads>0 ? remain_threads:1;			//如果这个数量不大于0，则把它置1

    if(removing_threads == 0)
        return pool->active_threads;

    //从id的最后一位线程开始删除
    for( i = pool->active_threads-1; i > remain_threads - 1; i--)	
    {
        errno = pthread_cancel(pool->tids[i]);
        DEBUG_RED("REMOVE:[%d] %u", i, pool->tids[i]);
        if(errno != 0)
            break;
    }

    if(i == pool->active_threads-1)		//判断是否取消掉要求的数量
    {
        return -1;
    }else
    {
        pool->active_threads = i+1;		//将新的线程数量登记active_threads
    }

    return pool->active_threads;		//返回剩下多少条线程在线程池中
}

bool destroy_pool(thread_pool **p)
{
    int i;
    thread_pool *pool = *p;

    pool->shutdown = true;					// 使各线程退出
    pthread_cond_broadcast(&pool->cond);	// 唤醒全部所有的线程

    for(i = 0; i<pool->active_threads; i++)	//开始接合线程
    {
        errno = pthread_join(pool->tids[i], NULL);
        if(errno != 0) {
            DEBUG_RED("join tids[%d] error: %s",	i, strerror(errno));
        } else {
            DEBUG_BLUE("destroy TID:[%u] ", (pthread_t)pool->tids[i]);
        }
    }
    //printf("任务完成。\n");
    free(pool->task_list);		//释放掉任务头节点
    free(pool->tids);			//释放掉线程ID内存
    free(pool);
    *p = NULL;
    return true;
}

